package com.xxx.flink.transform;

import com.xxx.flink.customsource.CustomSource;
import com.xxx.flink.pojo.Event;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 转换算子Rich富函数
 */
public class TransformRichTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据
        DataStreamSource<Event> sourceStream = env.addSource(new CustomSource());

        sourceStream.map(new MyRichMapper()).print();

        env.execute();
    }

    public static class MyRichMapper extends RichMapFunction<Event, Integer>{
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            System.out.println("open()函数被调用. taskName:" + getRuntimeContext().getTaskName());
        }

        @Override
        public Integer map(Event event) throws Exception {
            return event.getUrl().length();
        }

        @Override
        public void close() throws Exception {
            super.close();
            System.out.println("close()函数被调用.taskName:" + getRuntimeContext().getTaskName());
        }
    }
}
